package org.example.flinksql.operation;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * Set Operations
 * batch均支持 stream只支持Union、In
 */
public class Set {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
        TableEnvironment tbEnv = TableEnvironment.create(bsSettings);

        tbEnv.executeSql("CREATE TEMPORARY TABLE t_info_a (" +
                "id BIGINT," +
                "name STRING," +
                "PRIMARY KEY (id) NOT ENFORCED) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_info_a'," +
                "'username' = 'root'," +
                "'password' = '123456'" +
                ")");
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_info_b (" +
                "id BIGINT," +
                "name STRING," +
                "PRIMARY KEY (id) NOT ENFORCED) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_info_b'," +
                "'username' = 'root'," +
                "'password' = '123456'" +
                ")");

        Table a = tbEnv.sqlQuery("select * from t_info_a");
        Table b = tbEnv.sqlQuery("select * from t_info_b");

        // Union
//        a.union(b).execute().print();
//        a.unionAll(b).execute().print();

        // Intersect
//        a.intersect(b).execute().print();
//        a.intersectAll(b).execute().print();

        // Minus
//        a.minus(b).execute().print();
//        a.minusAll(b).execute().print();


        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
